/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.example.hrpc.rpc.server;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.apache.commons.lang3.SystemUtils;
import org.example.hrpc.common.utils.*;
import org.example.hrpc.rpc.message.Request;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import static org.example.hrpc.rpc.handler.Decoder.DECODER;
import static org.example.hrpc.rpc.handler.Encoder.ENCODER;

/**
 */
public class TransportServer implements Closeable {
    private static final Logger logger = LoggerFactory.getLogger(TransportServer.class);
    private final TransportServerHandler serverHandler;

    private ServerBootstrap bootstrap;
    private ChannelFuture channelFuture;
    private int port = -1;
    private final PooledByteBufAllocator pooledAllocator;


    /**
     * Creates a TransportServer that binds to the given host and the given port, or to any available
     * if 0. If you don't want to bind to any special host, set "hostToBind" to null.
     * */
    public TransportServer(
            String hostToBind,
            int portToBind,
            TransportServerHandler nettyServerHandler) {
        this.serverHandler = nettyServerHandler;
        this.pooledAllocator = NettyUtils.getSharedPooledByteBufAllocator(false, true);

        boolean shouldClose = true;
        try {
            init(hostToBind, portToBind);
            shouldClose = false;
        } finally {
            if (shouldClose) {
                JavaUtils.closeQuietly(this);
            }
        }
    }

    public int getPort() {
        if (port == -1) {
            throw new IllegalStateException("Server not initialized");
        }
        return port;
    }

    /**
     * 参考Netty的Server端程序开发实例
     * @param hostToBind
     * @param portToBind
     */
    private void init(String hostToBind, int portToBind) {

        IOMode ioMode = IOMode.NIO;
        EventLoopGroup bossGroup = NettyUtils.createEventLoop(ioMode, 1, "server-boss");
        int numThreads = Math.min(Runtime.getRuntime().availableProcessors() + 1, 32);
        EventLoopGroup workerGroup =  NettyUtils.createEventLoop(ioMode, numThreads, "server-worker");

        bootstrap = new ServerBootstrap()
                .group(bossGroup, workerGroup)
                .channel(NettyUtils.getServerChannelClass(ioMode))
                .option(ChannelOption.ALLOCATOR, pooledAllocator)
                .option(ChannelOption.SO_REUSEADDR, !SystemUtils.IS_OS_WINDOWS)
                .childOption(ChannelOption.ALLOCATOR, pooledAllocator);


        /**
         * 为根引导程序设置 pipeline初始化回调函数
         * bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
         *     @Override
         *     protected void initChannel(SocketChannel socketChannel) throws Exception {
         *         //给pipeline管道设置自定义的处理器
         *         socketChannel.pipeline().addLast(new MyServerHandler());
         *     }
         * });
         */
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                logger.debug("New connection accepted for remote address {}.", ch.remoteAddress());
                initializePipeline(ch, serverHandler);
            }
        });

        InetSocketAddress address = hostToBind == null ?
                new InetSocketAddress(portToBind): new InetSocketAddress(hostToBind, portToBind);
        channelFuture = bootstrap.bind(address);
        channelFuture.syncUninterruptibly();


        logger.debug("Shuffle server started on port: {}", port);
    }



    @Override
    public void close() {
        if (channelFuture != null) {
            // close is a local operation and should finish within milliseconds; timeout just to be safe
            channelFuture.channel().close().awaitUninterruptibly(10, TimeUnit.SECONDS);
            channelFuture = null;
        }
        if (bootstrap != null && bootstrap.config().group() != null) {
            bootstrap.config().group().shutdownGracefully();
        }
        if (bootstrap != null && bootstrap.config().childGroup() != null) {
            bootstrap.config().childGroup().shutdownGracefully();
        }
        bootstrap = null;
    }

    private void initializePipeline(SocketChannel ch, TransportServerHandler serverHandler) {
        try {
            /**
             * 对pipeline进行设置，这里的执行顺序参考Netty
             */
            ChannelPipeline pipeline = ch.pipeline()
                    .addLast("encoder", ENCODER)
                    .addLast("decoder", DECODER)
                    .addLast("idleStateHandler",
                            new IdleStateHandler(0, 0, connectionTimeoutMs() / 1000))
                    .addLast("handler", serverHandler);
        } catch (RuntimeException e) {
            logger.error("Error while initializing Netty pipeline", e);
            throw e;
        }
    }

    private int connectionTimeoutMs() {
        return 10_000;
    }


    public final static TransportServer createServer(String hostToBind, int portToBind) {
//        RequestHandler requestHandler = new RequestHandler();
//        TransportServerHandler serverHandler1 = new TransportServerHandler(requestHandler);
        TransportServerHandler serverHandler1 = new TransportServerHandler();
        return new TransportServer(hostToBind, portToBind, serverHandler1);
    }

}
